package com.example.concurrent;

import org.springframework.util.Assert;
import org.springframework.util.StopWatch;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import sun.font.CreatedFontTracker;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;

/**
 * @author houyangfan
 * @version 1.0
 * @date 2022/6/29 16:33
 *
 *
 */
@RestController
@RequestMapping("concurrentHashMap")
public class ConcurrentHashMapBug {

    // 总元素数量
    private static int ITEM_AMOUNT = 1000;
    // 线程个数
    private static int THREAD_AMOUNT = 10;

    public ConcurrentHashMap<String, Long> getData(int count){
        return LongStream.rangeClosed(1, count)
                .boxed()
                .collect(Collectors.toConcurrentMap(i -> UUID.randomUUID().toString(), Function.identity(),
                        (o1, o2) -> o1, ConcurrentHashMap::new));
    }

    @GetMapping("bugTest")
    public Integer testBug() throws InterruptedException {

        // 初始化map容器中的数据
        ConcurrentHashMap<String, Long> mapData = getData(ITEM_AMOUNT - 100);
        System.out.println("初始化数量："+ mapData.size());
        ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_AMOUNT);
        forkJoinPool.execute(()-> IntStream.rangeClosed(1,10).parallel().forEach(i -> {
            synchronized (mapData) {
                // 查询还需要多少个元素补充满
                int gap = ITEM_AMOUNT - mapData.size();
                System.out.println(Thread.currentThread().getName() + "线程需要补充gap ：" + gap);
                mapData.putAll(getData(gap));
            }
        }));
        //等待所有任务完成
        forkJoinPool.shutdown();
        System.out.println("finish size:"+ mapData.size());
        return 200;
    }

    @GetMapping("testMap")
    public Integer testMap() throws InterruptedException {
        AtomicInteger atomicInteger = new AtomicInteger();
        // 初始化map容器中的数据
        ConcurrentHashMap<String, Integer> mapData = new ConcurrentHashMap<>();
        System.out.println("初始化数量："+ mapData.size());
        ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_AMOUNT);
        forkJoinPool.execute(()-> IntStream.rangeClosed(1,10).parallel().forEach(i -> {
           mapData.put("num" + atomicInteger.get(),atomicInteger.get());
        }));
        //等待所有任务完成
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
        System.out.println("finish size:"+ mapData);
        return 200;
    }

    /**
     * 使用map来统计key出现次数的业务场景
     * @param loopCount 循环次数
     * @param threadCount 线程数量
     * @param itemCount 存入map中的元素数量
     * @return
     */
    public static Map<String, Long> normalUse(int loopCount, int threadCount, int itemCount) throws InterruptedException {

        ConcurrentHashMap<String, Long> concurrentHashMap = new ConcurrentHashMap<>(itemCount);
        ForkJoinPool forkJoinPool = new ForkJoinPool(threadCount);
        forkJoinPool.execute(() -> IntStream.rangeClosed(1, loopCount).parallel().forEach(i ->{
            // 获取随机key
            String key = "item" + ThreadLocalRandom.current().nextInt(itemCount);
            synchronized (concurrentHashMap) {
                if (concurrentHashMap.containsKey(key)){
                    //存在
                    concurrentHashMap.put(key, concurrentHashMap.get(key) + 1);
                } else {
                    concurrentHashMap.put(key, 1L);
                }
            }
        }));
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
        return concurrentHashMap;
    }

    public static Map<String, Long> goodUse(int loopCount, int threadCount, int itemCount) throws InterruptedException {

        ConcurrentHashMap<String, LongAdder> concurrentHashMap = new ConcurrentHashMap<>(itemCount);
        ForkJoinPool forkJoinPool = new ForkJoinPool(threadCount);
        forkJoinPool.execute(() -> IntStream.rangeClosed(1, loopCount).parallel().forEach(i ->{
            // 获取随机key
            String key = "item" + ThreadLocalRandom.current().nextInt(itemCount);
            concurrentHashMap.computeIfAbsent(key, k -> new LongAdder()).increment();
        }));
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
        return concurrentHashMap.entrySet().stream().collect(Collectors.toMap(e -> e.getKey(),e->e.getValue().longValue()));
    }

    public static void main(String[] args) {
        try {
            StopWatch stopWatch = new StopWatch();
            stopWatch.start("normalUse");
            Map<String,Long> nomalMap = normalUse(10000000, 10, 10);
            Assert.isTrue(nomalMap.size() == 1, "nomal size error");
            Assert.isTrue(nomalMap.entrySet().stream()
                            .mapToLong(item -> item.getValue()).reduce(0, Long::sum) == 10000000
                    , "normaluse count error");
            stopWatch.stop();
            stopWatch.start("goodUse");
            Map<String, Long> goodMap = goodUse(10000000, 10, 10);
            Assert.isTrue(goodMap.size() == 10, "nomal size error");
            Assert.isTrue(goodMap.entrySet().stream()
                            .mapToLong(item -> item.getValue()).reduce(0, Long::sum) == 10000000
                    , "normaluse count error");
            stopWatch.stop();
            System.out.println(stopWatch.prettyPrint());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
